package com.study.ProcessFunction;

import com.study.entity.OrderEvent;
import com.study.entity.PayEvent;
import com.study.utils.JsonHelper;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Date;
import java.util.Properties;

/**
 * @author sjw
 * @className Check
 * @date 2020/11/28 18:29
 * @description: TODO
 */
public class Check {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.10.200:9092");
        props.put("zookeeper.connect", "192.168.10.200:2181");
        props.put("group.id", "metric-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  //key 反序列化
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "latest");


        OutputTag orderEventTag = new OutputTag<String>("orderEventTag") {
        };
        OutputTag payEventTag = new OutputTag<String>("payEventTag") {
        };

        KeyedStream<OrderEvent, String> orderEventStringKeyedStream = getOrderEventStringKeyedStream(env, props);
        //orderEventStringKeyedStream.print();

        KeyedStream<PayEvent, String> payEventStringKeyedStream = getOrPayEventStringKeyedStream(env, props);
        //payEventStringKeyedStream.print();
        SingleOutputStreamOperator<String> process =
                orderEventStringKeyedStream.connect(payEventStringKeyedStream).process(new MyCoProcessFunction());
        process.print("success");
        process.getSideOutput(orderEventTag).print("orderEventTag");
        process.getSideOutput(payEventTag).print("payEventTag");

        env.execute();
    }

    private static KeyedStream<OrderEvent, String> getOrderEventStringKeyedStream(StreamExecutionEnvironment env, Properties props) {
        return env.addSource(new FlinkKafkaConsumer010<String>(
                "orderEvent", new SimpleStringSchema(), props)).map(new MapFunction<String, OrderEvent>() {
            @Override
            public OrderEvent map(String s) throws Exception {
                return JsonHelper.jsonToObject(s, OrderEvent.class);
            }
        }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<OrderEvent>() {
            @Override
            public long extractAscendingTimestamp(OrderEvent element) {
                return element.getTimestamp();
            }

        }).filter(new FilterFunction<OrderEvent>() {
            @Override
            public boolean filter(OrderEvent orderEvent) throws Exception {
                return "pay".equals(orderEvent.getAction());
            }
        }).keyBy(new KeySelector<OrderEvent, String>() {
            @Override
            public String getKey(OrderEvent orderEvent) throws Exception {
                return orderEvent.getOrderId();
            }
        });
    }


    private static KeyedStream<PayEvent, String> getOrPayEventStringKeyedStream(StreamExecutionEnvironment env, Properties props) {
        return env.addSource(new FlinkKafkaConsumer010<String>(
                "payEvent", new SimpleStringSchema(), props)).map(new MapFunction<String, PayEvent>() {
            @Override
            public PayEvent map(String s) throws Exception {
                return JsonHelper.jsonToObject(s, PayEvent.class);
            }
        }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<PayEvent>() {
            @Override
            public long extractAscendingTimestamp(PayEvent element) {
                return element.getTimestamp();
            }
        }).keyBy(new KeySelector<PayEvent, String>() {
            @Override
            public String getKey(PayEvent payEvent) throws Exception {
                return payEvent.getOrderId();
            }
        });
    }

    private static class MyCoProcessFunction extends CoProcessFunction<OrderEvent, PayEvent, String> {

        OutputTag orderEventTag = new OutputTag<String>("orderEventTag") {
        };
        OutputTag payEventTag = new OutputTag<String>("payEventTag") {
        };

        ValueState<OrderEvent> orderEventValueState = null;
        ValueState<PayEvent> payEventValueState = null;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            // 状态的注册
            ValueStateDescriptor<OrderEvent> orderEventValueStateDescriptor = new ValueStateDescriptor<>(
                    "OrderEvent-state",
                    OrderEvent.class);
            orderEventValueState = getRuntimeContext().getState(orderEventValueStateDescriptor);

            ValueStateDescriptor<PayEvent> payEventValueStateDescriptor = new ValueStateDescriptor<>(
                    "PayEvent-state",
                    PayEvent.class);
            payEventValueState = getRuntimeContext().getState(payEventValueStateDescriptor);
        }

        @Override
        public void processElement1(OrderEvent orderEvent, Context ctx, Collector<String> out) throws Exception {
            System.out.println("processElement1  "+new Date(orderEvent.getTimestamp())+"  "+JsonHelper.objectToJson(orderEvent));
            PayEvent payEvent = payEventValueState.value();
            if (payEvent != null) {
                out.collect("订单事件： " + orderEvent.toString() + " 和交易事件： " + payEvent.toString());
                payEventValueState.clear();
                orderEventValueState.clear();
            } else {
                orderEventValueState.update(orderEvent);
                ctx.timerService().registerEventTimeTimer(orderEvent.getTimestamp() + 5000L);
                System.out.println("processElement1 设置定时器 " + new Date(orderEvent.getTimestamp() + 5000L));
            }
        }

        @Override
        public void processElement2(PayEvent payEvent, Context ctx, Collector<String> out) throws Exception {
            System.out.println("processElement2  "+new Date(payEvent.getTimestamp())+"  "+JsonHelper.objectToJson(payEvent));
            OrderEvent orderEvent = orderEventValueState.value();
            if (orderEvent != null) {
                out.collect("订单事件： " + orderEvent.toString() + " 和交易事件： " + payEvent.toString());
                payEventValueState.clear();
                orderEventValueState.clear();
            } else {
                payEventValueState.update(payEvent);
                ctx.timerService().registerEventTimeTimer(payEvent.getTimestamp() + 3000L);
                System.out.println("processElement2 设置定时器 " + new Date(payEvent.getTimestamp() + 3000L));
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            System.out.println("onTimer "+timestamp);
            if (orderEventValueState.value() == null || payEventValueState.value() == null) {
                if (orderEventValueState.value() != null) {
                    ctx.output(orderEventTag, orderEventValueState.value().toString() + " 有order事件没有pay事件，属于异常事件");
                }
                if (payEventValueState.value() != null) {
                    ctx.output(payEventTag, payEventValueState.value().toString() + " 有pay事件没有order事件，属于异常事件");
                }
            } else {
                orderEventValueState.clear();
                payEventValueState.clear();
            }
        }


    }
}
